from ipaddress import v6_int_to_packed
from feapder.pipelines import BasePipeline
from typing import Dict, List, Tuple
import feapder.setting as setting
from kafka import KafkaProducer
from feapder.utils import tools
from datetime import datetime


class KafkaPipeline(BasePipeline):
    """
    pipeline 是单线程的，批量保存数据的操作，不建议在这里写网络请求代码，如下载图片等
    """
    def __init__(self):
        self._to_db = None

    @property
    def to_db(self):
        if not self._to_db:
            self._to_db = KafkaProducer(bootstrap_servers=setting.BOOTSTRAP_SERVERS, max_request_size=2097152)

        return self._to_db

        
    def save_items(self, table, items: List[Dict]) -> bool:
        """
        保存数据
        Args:
            table: 表名
            items: 数据，[{},{},...]

        Returns: 是否保存成功 True / False
                 若False，不会将本批数据入到去重库，以便再次入库

        """

        print("自定义pipeline save_items， 保存数据 >>>>", table, items)

        return True

    def update_items(self, table, items: List[Dict], update_keys=Tuple) -> bool:
        """
        更新数据, 与UpdateItem配合使用，若爬虫中没使用UpdateItem，则可不实现此接口
        Args:
            table: 表名
            items: 数据，[{},{},...]
            update_keys: 更新的字段, 如 ("title", "publish_time")

        Returns: 是否更新成功 True / False
                 若False，不会将本批数据入到去重库，以便再次入库

        """

        print("自定义pipeline,update_items 更新数据 >>>>", table, items, update_keys)
        for item in items:
            topic = setting.TOPIC_MAP.get(table)  #kafka topic 映射
            if topic:
                [item.update({k:v.strftime("%Y-%m-%d %H:%M:%S")}) for k,v in item.items() if isinstance(v,datetime)]
                
                future = self.to_db.send(topic=topic,value=tools.json.dumps(item).encode())
                result = future.get(timeout=60)
                kafka_log = {
                    'topic': result.topic,
                    'partition': result.partition,
                    'offset': result.offset,
                    'log_start_offset': result.log_start_offset,
                }
                print (f"push kafka success: {kafka_log} ")